package com.grf.rocketmq;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * @author : guorf
 * @version : v1.0
 * @description : TODO 拉取消息
 * @createTime : 2023/9/13 16:34
 */
@Slf4j
public class PullConsumer {
    @SneakyThrows
    public static void main(String[] args) {
//        oldPull();
//        litePullConsumerSubscribe();
        litePullConsumerAssign();
    }

    @SneakyThrows
    public static void oldPull(){
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(ProducerUtil.GROUP_NAME);
        consumer.setNamesrvAddr(ProducerUtil.NAMESRV_ADDR);
        consumer.start();
        try {
            MessageQueue mq = new MessageQueue();
            mq.setQueueId(0);
            mq.setTopic(ProducerUtil.TOPIC);
            mq.setBrokerName("master-broker-a");
            long offset = 26;
            PullResult pullResult = consumer.pull(mq, "*", offset, 32);
            if (pullResult.getPullStatus().equals(PullStatus.FOUND)) {
//                System.out.printf("%s%n", pullResult.getMsgFoundList());
                List<MessageExt> msgs = pullResult.getMsgFoundList();
                for (MessageExt msg: msgs) {
                    log.info("topic is: {}, tags is:{}, queue is: {}, body is: {}, property is: {}", msg.getTopic(), msg.getTags(), msg.getQueueId(), new String(msg.getBody(),"UTF-8"), msg.getUserProperty("user"));
                }
                consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        consumer.shutdown();
    }

    public static volatile boolean running = true;
    /**
     * @description TODO 订阅方式拉取
     * LitePullConsumer拉取消息调用的是轮询poll接口，如果能拉取到消息则返回对应的消息列表，否则返回null
     **/
    @SneakyThrows
    public static void litePullConsumerSubscribe(){
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(ProducerUtil.GROUP_NAME);
        litePullConsumer.setNamesrvAddr(ProducerUtil.NAMESRV_ADDR);
        litePullConsumer.subscribe(ProducerUtil.TOPIC, "*");
        litePullConsumer.setPullBatchSize(20);
        litePullConsumer.start();
        try {
            while (running) {
                int index = 0;
                List<MessageExt> messageExts = litePullConsumer.poll();
//                System.out.printf("%s%n", messageExts);
                for (MessageExt msg: messageExts) {
                    log.info("index is: {},topic is: {}, tags is:{}, queue is: {}, body is: {}, property is: {}", index, msg.getTopic(), msg.getTags(), msg.getQueueId(), new String(msg.getBody(),"UTF-8"), msg.getUserProperty("user"));
                    index++;
                }
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }

    /**
     * @description TODO Assign模式
     * Assign模式下没有自动的负载均衡机制，需要用户自行指定需要拉取的队列
     **/

    @SneakyThrows
    public static void litePullConsumerAssign(){
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(ProducerUtil.GROUP_NAME);
        litePullConsumer.setNamesrvAddr(ProducerUtil.NAMESRV_ADDR);
        // 设置非自动提交
        litePullConsumer.setAutoCommit(false);
        litePullConsumer.start();
        Collection<MessageQueue> mqSet = litePullConsumer.fetchMessageQueues(ProducerUtil.TOPIC);
        List<MessageQueue> list = new ArrayList<>(mqSet);
        List<MessageQueue> assignList = new ArrayList<>();
        for (int i = 0; i < list.size() / 2; i++) {
            assignList.add(list.get(i));
        }
        // 设置拉取队列
        litePullConsumer.assign(assignList);
//        设置第一个队列拉取的位点从10开始
        litePullConsumer.seek(assignList.get(0), 10);
        try {
            while (running) {
                int index = 0;
                List<MessageExt> messageExts = litePullConsumer.poll();
//                System.out.printf("%s %n", messageExts);
                for (MessageExt msg: messageExts) {
                    log.info("index is: {},topic is: {}, tags is:{}, queue is: {}, body is: {}, property is: {}", index, msg.getTopic(), msg.getTags(), msg.getQueueId(), new String(msg.getBody(),"UTF-8"), msg.getUserProperty("user"));
                    index++;
                }
                litePullConsumer.commitSync();
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }

}
